package com.codejiwei.flink.zk;

import org.apache.flink.shaded.curator4.org.apache.curator.RetryPolicy;
import org.apache.flink.shaded.curator4.org.apache.curator.framework.CuratorFramework;
import org.apache.flink.shaded.curator4.org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.flink.shaded.curator4.org.apache.curator.retry.ExponentialBackoffRetry;

import java.nio.charset.StandardCharsets;
import java.util.Properties;
import java.util.concurrent.atomic.AtomicReference;

/**
 * author: codejiwei
 * date: 2023/9/7
 * desc:
 **/
public class ZookeeperUtil {

    private static AtomicReference<CuratorFramework> clientReference = new AtomicReference<>();
    public static CuratorFramework getZkClient(Properties properties) {
        if (clientReference.get() == null) {
            clientReference.set(connect(properties));
        }
        return clientReference.get();
    }

    public static CuratorFramework connect(Properties properties) {
        ExponentialBackoffRetry retryPolicy = new ExponentialBackoffRetry(3000, 3);
        return connect(properties.getProperty("address"),
                Integer.parseInt(properties.getProperty("sessionTimeout")),
                Integer.parseInt(properties.getProperty("connectTimeout")),
                properties.getProperty("prefix"),
                retryPolicy
        );
    }


    private static CuratorFramework connect(String address, Integer sessionTimeout, Integer connectTimeout, String prefix, RetryPolicy retryPolicy) {
        return CuratorFrameworkFactory.builder()
                .namespace(prefix)
                .connectString(address)
                .sessionTimeoutMs(sessionTimeout)
                .connectionTimeoutMs(connectTimeout)
                .retryPolicy(retryPolicy)
                .build();
    }

    public static String getData(CuratorFramework client, String path) throws Exception {
        return new String (client.getData().forPath(path), StandardCharsets.UTF_8);
    }


}
